package org.apache.flink;

import org.apache.flink.sink.ProducerStringSerializationSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.ArrayList;
import java.util.Properties;

public class KafkaSinkApp {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop000:9092");

        String topic = "cmak_test";
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(topic,
                new ProducerStringSerializationSchema(topic),
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);

        ArrayList<String> list = new ArrayList<>();
        list.add(1+"");
        list.add(2+"");
        list.add(3+"");
        env.fromCollection(list).addSink(producer).setParallelism(1);

        env.execute("sink");
    }
}
